package saramax

import (
	"bbs-web/pkg/logger"
	"context"
	"encoding/json"
	"github.com/IBM/sarama"
	"time"
)

type BatchHandler[T any] struct {
	l  logger.Logger
	fn func(msg []*sarama.ConsumerMessage, t []T) error
}

func NewBatchHandler[T any](l logger.Logger, fn func(msg []*sarama.ConsumerMessage, t []T) error) *BatchHandler[T] {
	return &BatchHandler[T]{l: l, fn: fn}
}

func (b *BatchHandler[T]) Setup(session sarama.ConsumerGroupSession) error {
	return nil
}

func (b *BatchHandler[T]) Cleanup(session sarama.ConsumerGroupSession) error {
	return nil
}

func (b *BatchHandler[T]) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	const batchSize = 10
	msgs := claim.Messages()
	for {
		batch := make([]*sarama.ConsumerMessage, 0, batchSize)
		ts := make([]T, 0, batchSize)
		ctx, cancle := context.WithTimeout(context.Background(), time.Second)
		done := false
		for i := 0; i < batchSize && !done; i++ {
			select {
			case <-ctx.Done():
				done = true

			case msg, ok := <-msgs:
				if !ok {
					cancle()
					return nil
				}
				var t T
				//因为Unmarshal函数需要将解析后的JSON数据填充到目标数据结构中，而只有通过指针传递才能够修改目标数据结构的值。
				err := json.Unmarshal(msg.Value, &t)
				if err != nil {
					b.l.Error("反序列化消息失败",
						logger.Error(err),
						logger.Int64("Offset", msg.Offset),
						logger.Int32("Partition", msg.Partition),
						logger.String("Topic", msg.Topic),
						logger.String("Value", string(msg.Value)),
					)
					continue
				}
				batch = append(batch, msg)
				ts = append(ts, t)
			}
		}
		cancle()
		// 凑够了一批，使用fn函数处理
		err := b.fn(batch, ts)
		if err != nil {
			b.l.Error("处理消息失败",
				// 把真个 msgs 都记录下来
				logger.Error(err))
		}

		//在 Kafka 中，消费者处理完消息后需要向 Kafka 提交确认，告知 Kafka 消息已被处理。这个过程通常称为消息提交。
		for _, msg := range batch {
			session.MarkMessage(msg, "")
		}

	}

}
